package com.iflytek.iot.common.log.log2kafka.log;

import com.iflytek.iot.common.log.log2kafka.model.LogModel;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author zurun
 * @date 2020/8/11 19:06:25
 */
public class Log2kafkaQueue {
    /**
     * 写日志失败的
     */
    private static final AtomicLong addedFailed = new AtomicLong(0L);
    private static final AtomicInteger addCount = new AtomicInteger(0);
    private static final AtomicInteger takeCount = new AtomicInteger(0);
    /**
     * 忽略的
     */
    private static final AtomicLong ignoreCount = new AtomicLong(0);
    /**
     * 队列长度,1<<18为26万
     */
    private static final BlockingQueue<LogModel> queue = new ArrayBlockingQueue(1 << 18);
    private static final AtomicInteger index = new AtomicInteger(0);

    static {
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
            int add = addCount.getAndSet(0);
            int take = takeCount.getAndSet(0);
            if (add > 0) {
                System.out.println(System.currentTimeMillis() +
                        " queue:" + formatNum(queue.size(), 3) +
                        " ; add:" + formatNum(add, 4) +
                        " ; take:" + formatNum(take, 4) +
                        " ; w2k:" + LogConsumer2Kafka.sum.get() +
                        " ; f:" + addedFailed.get() +
                        " ; ignore:" + ignoreCount.get())
                ;
            }
        }, 0, 1, TimeUnit.SECONDS);
    }

    public static void incrementIgnore() {
        ignoreCount.incrementAndGet();
    }

    public static boolean add(LogModel logBean) {
        int num = index.getAndIncrement();
        if (num >= 9_9999) {
            while (true) {
                int t = index.get();
                if (t >= 9_9999) {
                    boolean b = index.compareAndSet(t, 0);
                    if (b) {
                        break;
                    }
                } else {
                    break;
                }
            }
        }

        logBean.setLogIndex(num);
        boolean bl = queue.offer(logBean);
        addCount.getAndIncrement();
        if (!bl) {
            addedFailed.getAndIncrement();
        }
        return bl;
    }

    public static LogModel take(long time, TimeUnit timeUnit) throws InterruptedException {
        LogModel logBean = queue.poll(time, timeUnit);
        if (logBean != null) {
            takeCount.getAndIncrement();
        }
        return logBean;
    }

    public static int queueSize() {
        return queue.size();
    }

    public static void main(String[] args) {
        System.out.println(1 << 18);
        ExecutorService executorService = Executors.newFixedThreadPool(100);
        for (int i = 0; i < 100; i++) {
            executorService.execute(() -> {
                while (true) {
                    int num = index.getAndIncrement();
                    if (num > 9_0) {
                        while (true) {
                            int t = index.get();
                            if (t > 9_0) {
                                boolean b = index.compareAndSet(t, 0);
                                if (b) {
                                    break;
                                }
                            } else {
                                break;
                            }
                        }
                    }
                    System.out.println(num);
                }
            });
        }

    }

    private static String formatNum(int num, int length) {

        return String.format("%" + length + "d", num);
    }

    private static String formatNum(long num, int length) {

        return String.format("%" + length + "d", num);
    }


}
